package com.zdb.demo.flink.streaming;

import org.apache.commons.io.IOUtils;
import org.apache.flink.api.common.functions.*;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.io.File;
import java.io.FileReader;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

/**
 * 实现一个 Flink Batch Job
 *
 * 使用 Broadcast 将字典广播，字典包含数据编号（视为商品编号）到另外一个数据编号（视为厂家编号）的映射
 * 使用 Distributed Cache 将厂家详细信息文件广播。该文件每行包含三个字段，用逗号分隔，第一个字段是厂家编号，第二个字段是厂家名称，第三个字段是厂家城市名
 * 最后需要按城市 group by，然后对商品个数做 count
 */
public class H7 {

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration(){{
            setInteger("rest.port", 9191);
            setBoolean("local.start-webserver", true);
        }};
        final ExecutionEnvironment batchEnv = ExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);

        DataSet<Tuple2<String, String>> broadcastDS = batchEnv.fromElements(new Tuple2<String, String>("i1", "f1"), new Tuple2<String, String>("i2", "f1"), new Tuple2<String, String>("i3", "f2"), new Tuple2<String, String>("i4", "f2"));

        // factory_id,factory_name, factory_city
        batchEnv.registerCachedFile("D:\\workspace\\flink-demo\\flink-streaming\\data\\factoryinfo.csv", "factory_info");

        batchEnv.fromElements(new Tuple1<String>("i1"), new Tuple1<>("i2"), new Tuple1<>("i3"))
                .map(new RichMapFunction<Tuple1<String>, Tuple2<String, String>>() {    // item_id, factory_city

                    private Map<String, String> factoryCityMap = new HashMap<>();
                    private Map<String, String> item2factory = new HashMap<>();

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //
                        File file = getRuntimeContext().getDistributedCache().getFile("factory_info");
                        IOUtils.readLines(new FileReader(file)).forEach(line -> {
                            String[] strs = line.split(",");
                            factoryCityMap.put(strs[0], strs[2]);
                        });
                        // broadcast
                        Collection<Tuple2<String, String>> broadcastMap = getRuntimeContext().getBroadcastVariable("item2factory");
                        broadcastMap.forEach(tuple2 -> item2factory.put(tuple2.f0, tuple2.f1));
                    }

                    @Override
                    public Tuple2<String, String> map(Tuple1<String> value) throws Exception {

                        String city = factoryCityMap.get(item2factory.get(value.f0));

                        return new Tuple2<>(value.f0, city);
                    }
                }).withBroadcastSet(broadcastDS, "item2factory").groupBy(1)
                .reduceGroup((Iterable<Tuple2<String, String>> tuples, Collector<Tuple2<String, Integer>> out) -> {
                    int count = 0;
                    String city = null;
                    for(Tuple2<String, String> tuple : tuples) {
                        count += 1;
                        city = tuple.f1;
                    }
                    out.collect(new Tuple2<>(city, count));
                }).returns(Types.TUPLE(Types.STRING, Types.INT)).print();

    }

}
